package com.example.rocketmq.controller;

import com.example.rocketmq.utils.ListSplitter;
import com.example.rocketmq.utils.SendCallbackListener;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.TreeMap;

@RestController
public class ProducerController {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    private static final List<Integer> DELAY_TIME_LEVEL = new ArrayList<>();
    static {
        // 预设值的延迟时间间隔为：1s、 5s、 10s、 30s、 1m、 2m、 3m、 4m、 5m、 6m、 7m、 8m、 9m、 10m、 20m、 30m、 1h、 2h
        // 对应1-18个等级
        DELAY_TIME_LEVEL.add(1);
        DELAY_TIME_LEVEL.add(5);
        DELAY_TIME_LEVEL.add(10);
        DELAY_TIME_LEVEL.add(30);
        DELAY_TIME_LEVEL.add(60);
        DELAY_TIME_LEVEL.add(60 * 2);
        DELAY_TIME_LEVEL.add(60 * 3);
        DELAY_TIME_LEVEL.add(60 * 4);
        DELAY_TIME_LEVEL.add(60 * 5);
        DELAY_TIME_LEVEL.add(60 * 6);
        DELAY_TIME_LEVEL.add(60 * 7);
        DELAY_TIME_LEVEL.add(60 * 8);
        DELAY_TIME_LEVEL.add(60 * 9);
        DELAY_TIME_LEVEL.add(60 * 10);
        DELAY_TIME_LEVEL.add(60 * 20);
        DELAY_TIME_LEVEL.add(60 * 30);
        DELAY_TIME_LEVEL.add(60 * 60);
        DELAY_TIME_LEVEL.add(60 * 60 * 2);
    }

    @RequestMapping("/producer/{id}")
    public String producer(@PathVariable("id") int id) {
        rocketMQTemplate.convertAndSend("producer-order", "hello world:" + id);
        return "success";
    }

    /**
     * 发送同步消息
     *
     * @return
     */
    @RequestMapping("/test2")
    public SendResult producer() {
        int num = 2;
        Message<String> message = MessageBuilder.withPayload("hello world" + num).setHeader(RocketMQHeaders.KEYS, num).build();
        SendResult sendResult = rocketMQTemplate.syncSend("producer-sync", message);
        return sendResult;
    }

    /**
     * 发送异步消息
     *
     * @return
     */
    @RequestMapping("/test3")
    public String producer3() {
        int num = 3;
        Message<String> message = MessageBuilder.withPayload("hello world" + num).setHeader(RocketMQHeaders.KEYS, num).build();

        rocketMQTemplate.asyncSend("producer-async", message, new SendCallbackListener(3));
        return "success";
    }

    /**
     * 发送延迟消息
     *
     * @return
     */
    @RequestMapping("/test4")
    public SendResult producer4() {
        int num = 4;
        int delay = 5;
        Message<String> message = MessageBuilder.withPayload("hello world" + num).setHeader(RocketMQHeaders.KEYS, num).build();
        System.out.println("hello world" + num+" "+"时间:"+new Date());
        SendResult sendResult = rocketMQTemplate.syncSend("producer-yc", message, 5000, getNearDelayLevel(delay));
        return sendResult;
    }

    /**
     * 发送批量消息
     *
     * @return
     */
    @RequestMapping("/test5")
    public SendResult producer5() {
        int num = 5;
        List<Message> messages = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            Message<String> message = MessageBuilder.withPayload("hello world" + i).setHeader(RocketMQHeaders.KEYS, num).build();
            messages.add(message);
        }
        // 批量下发消息到broker,不支持消息顺序操作，并且对消息体有大小限制（不超过4M）
        ListSplitter splitter = new ListSplitter(messages, 1000*1000*4);
        while (splitter.hasNext()) {
            List<Message> listItem = splitter.next();
            rocketMQTemplate.syncSend("topic-list", listItem);
        }
        SendResult sendResult = rocketMQTemplate.syncSend("producer-order", messages);
        return sendResult;
    }

    /**
     * 发送事务消息
     *
     * @return
     */
    @RequestMapping("/test6/{id}")
    public SendResult producer6(@PathVariable("id") int id) {
        int num = 6;
        Message<String> message = MessageBuilder.withPayload("hello world" + num)
                .setHeader(RocketMQHeaders.KEYS, num)
                .setHeader(RocketMQHeaders.TRANSACTION_ID, id)
                .build();
        SendResult sendResult = rocketMQTemplate.sendMessageInTransaction("producer-transaction", message, 1);
        return sendResult;
    }

    /**
     * 获取最接近的节点值
     **/
    private int getNearDelayLevel(int delay) {

        // 如果包含则直接返回
        if (DELAY_TIME_LEVEL.contains(delay)) {
            int index = DELAY_TIME_LEVEL.indexOf(delay);
            return index + 1;
        }

        //两个时间的绝对值 - 位置
        TreeMap<Integer, Integer> resultMap = new TreeMap<>();
        DELAY_TIME_LEVEL.forEach(time -> resultMap.put(Math.abs(delay - time), DELAY_TIME_LEVEL.indexOf(time) + 1));
        return resultMap.firstEntry().getValue();
    }
}
